package com.gzyj.video.api;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.gzyj.data.common.ResponseMsg;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.*;

import java.text.SimpleDateFormat;
import java.util.Date;

@RestController
@RequestMapping("api/video")
public class VideoApi {


    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;


    @Value("${kafka.queue.down}")
    private String queueDown;

    @Value("${video.server.ip}")
    private String videoServerIp;


    @GetMapping("play")
    public ResponseMsg<String> play(@RequestParam("tag") String tag, @RequestParam("bitstream") String bitstream) {

        String[] splits = tag.split("-");

        JSONObject param = new JSONObject();
        param.put("cmdID", 0x9101);
        param.put("ipAddress", splits[0]);

        param.put("encIpAdress", videoServerIp);
        param.put("encPortTCP", 1078);
        param.put("encPortUDP", 1078);
        param.put("channel", Byte.valueOf(splits[1]));
        param.put("dataType", Byte.valueOf("0"));
        param.put("bitstreamType", Byte.valueOf(bitstream));

        param.put("seq", 1);
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        param.put("sendTime", sdf.format(new Date()));


        kafkaTemplate.send(queueDown, JSON.toJSONString(param, SerializerFeature.DisableCircularReferenceDetect));

        return new ResponseMsg<>(200, "success");

    }


    @GetMapping("stop")
    public ResponseMsg<String> stop(@RequestParam("tag") String tag, @RequestParam("bitstream") String bitstream) {

        String[] splits = tag.split("-");

        JSONObject param = new JSONObject();
        param.put("cmdID", 0x9102);
        param.put("ipAddress", splits[0]);

        param.put("channel", Byte.valueOf(splits[1]));
        param.put("command", Byte.valueOf("0"));
        param.put("turnOffType", Byte.valueOf("0"));
        param.put("bitstreamType", Byte.valueOf(bitstream));

        param.put("seq", 1);
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        param.put("sendTime", sdf.format(new Date()));


        kafkaTemplate.send(queueDown, JSON.toJSONString(param, SerializerFeature.DisableCircularReferenceDetect));

        return new ResponseMsg<>(200, "success");

    }


}
